<head>
    <meta charset='utf-8'>
</head>

<ul>
    <li>
        <h2>正确理解其含义</h2>

        <ul>
            <li>500万，是指500万条聊天消息，每条消息包括groupid、usernick、userid、chattype、content、time等信息</li>
            <li>测试过程中server和client是位于同一台机器</li>
            <li>需要有SSD、i7这样的电脑配置</li>
            <li>需要拿t-io 1.6.9分支进行测试，由于众所周知的DDos攻击事件，t-io 1.7.0加入了链路跟踪、反攻击等功能，对性能有一定损耗</li>
            <li>不是每次测试都能达到500万/秒，可以理解为这是数次测试中取的一个峰值之一（注意此处的之一，不带之一的峰值可能还要大于这个500万），所以平均值是在300-400万之间，当然还需要去掉预热的前几次测试（预热时速度会有点慢，比平均值会少约30-60%的性能）</li>
            <li>这是异步收发的性能，同步的收发性能本人测试过约为20万</li>
        </ul>

    </li>
    <li>
        <h2>了解决定网络io性能的几大件</h2>

        <ul>
            <li>所依赖的io事件底层架构：t-io依赖jdk的aio，这为t-io垫定了一个比较好的基础</li>
            <li>线程模型：t-io在jdk线程池的基础上自行创造了一个同步安全的基于队列批量处理的Runnable及保证该Runnable被同步安全调度的Executer，把线程切换消耗几乎降到了零； 同时t-io中大量运用读写锁把资源维护所带来的同步消耗降到最低----这里友情提示一下，t-io性能的最大秘密就在这了
            </li>
            <li>编码解码：t-io灵活运用了jdk自带的bytebuffer，尽量复用已有的bytebuffer，为编码解码性能打下了基础，但是如果业务层的解码编码性能上不来，作为框架是无能为力的</li>
        </ul>
    </li>
    <li>
        <h2>t-io接收消息及处理流程</h2>

        <ul>
            <li>
                <h3>监听消息</h3>
            </li>
        </ul>

        <pre>
<code class="language-java">
public void start(String serverIp, int serverPort) throws IOException {
    this.serverNode = new Node(serverIp, serverPort);
    //		ExecutorService groupExecutor = serverGroupContext.getGroupExecutor();

    AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withThreadPool(serverGroupContext.getGroupExecutor());
    serverSocketChannel = AsynchronousServerSocketChannel.open(channelGroup);

    serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 64 * 1024);

    InetSocketAddress listenAddress = null;

    if (StringUtils.isBlank(serverIp)) {
        listenAddress = new InetSocketAddress(serverPort);
    } else {
        listenAddress = new InetSocketAddress(serverIp, serverPort);
    }

    serverSocketChannel.bind(listenAddress, 0);

    AcceptCompletionHandler<SessionContext, P, R> acceptCompletionHandler = serverGroupContext.getAcceptCompletionHandler();
    serverSocketChannel.accept(this, acceptCompletionHandler);

    log.warn("t-io server started, listen on {}", this.serverNode);
}
    </code></pre>

        <ul>
            <li>
                <h3>定义业务消息包结构</h3>
            </li>
        </ul>

        <p>注：有时候服务器和客户端的业务消息包结构不一样，这种情况下，消息包的定义就不要放在公共模块</p>

        <pre>
<code class="language-java">public class HelloPacket extends Packet
{
    public static final int HEADER_LENGHT = 4;//消息头的长度
    public static final String CHARSET = "utf-8";
    private byte[] body;
 
    /**
     * @return the body
     */
    public byte[] getBody()
    {
        return body;
    }
 
    /**
     * @param body the body to set
     */
    public void setBody(byte[] body)
    {
        this.body = body;
    }
}
           </code></pre>

        <ul>
            <li>
                <h3>实现io通讯中必不可少的编码和解码</h3>
            </li>
        </ul>

        <p>注：有时候服务器端和客户端的编码解码算法不一样，这个时候就不要以公共模块中完成编码和解码的代码，而是分别放在服务器端和客户端完成。</p>

        <pre>
<code class="language-java">/**
 * hello world版中服务器端和客户端的编码解码算法是一样的，所以抽象一个公共的父类出来
 * @author tanyaowu 
 *
 */
public abstract class HelloAbsAioHandler implements AioHandler&lt;Object, HelloPacket, Object&gt;
{
    /**
     * 编码：把业务消息包编码为可以发送的ByteBuffer
     * 总的消息结构：消息头 + 消息体
     * 消息头结构：    4个字节，存储消息体的长度
     * 消息体结构：   对象的json串的byte[]
     */
    @Override
    public ByteBuffer encode(HelloPacket packet, GroupContext&lt;Object, HelloPacket, Object&gt; groupContext, ChannelContext&lt;Object, HelloPacket, Object&gt; channelContext)
    {
        byte[] body = packet.getBody();
        int bodyLen = 0;
        if (body != null)
        {
            bodyLen = body.length;
        }
 
        //bytebuffer的总长度是 = 消息头的长度 + 消息体的长度
        int allLen = HelloPacket.HEADER_LENGHT + bodyLen;
        //创建一个新的bytebuffer
        ByteBuffer buffer = ByteBuffer.allocate(allLen);
        //设置字节序
        buffer.order(groupContext.getByteOrder());
 
        //写入消息头----消息头的内容就是消息体的长度
        buffer.putInt(bodyLen);
 
        //写入消息体
        if (body != null)
        {
            buffer.put(body);
        }
        return buffer;
    }
 
    /**
     * 解码：把接收到的ByteBuffer，解码成应用可以识别的业务消息包
     * 总的消息结构：消息头 + 消息体
     * 消息头结构：    4个字节，存储消息体的长度
     * 消息体结构：   对象的json串的byte[]
     */
    @Override
    public HelloPacket decode(ByteBuffer buffer, ChannelContext&lt;Object, HelloPacket, Object&gt; channelContext) throws AioDecodeException
    {
        int readableLength = buffer.limit() - buffer.position();
        //收到的数据组不了业务包，则返回null以告诉框架数据不够
        if (readableLength &lt; HelloPacket.HEADER_LENGHT)
        {
            return null;
        }
 
        //读取消息体的长度
        int bodyLength = buffer.getInt();
 
        //数据不正确，则抛出AioDecodeException异常
        if (bodyLength &lt; 0)
        {
            throw new AioDecodeException("bodyLength [" + bodyLength + "] is not right, remote:" + channelContext.getClientNode());
        }
 
        //计算本次需要的数据长度
        int neededLength = HelloPacket.HEADER_LENGHT + bodyLength;
        //收到的数据是否足够组包
        int isDataEnough = readableLength - neededLength;
        // 不够消息体长度(剩下的buffe组不了消息体)
        if (isDataEnough &lt; 0)
        {
            return null;
        } else  //组包成功
        {
            HelloPacket imPacket = new HelloPacket();
            if (bodyLength &gt; 0)
            {
                byte[] dst = new byte[bodyLength];
                buffer.get(dst);
                imPacket.setBody(dst);
            }
            return imPacket;
        }
    }
}</code></pre>

        <ul>
            <li>
                <h3>定义一下双方约定的端口</h3>
            </li>
        </ul>

        <pre>
<code class="language-java">public interface Const
{
    public static final int PORT = 6789;
}</code></pre>
    </li>
    <li>
        <h2>服务端代码</h2>

        <ul>
            <li>
                <h3>实现AioHandler</h3>
            </li>
        </ul>

        <p>注：此处继承自HelloAbsAioHandler，父类已经完成编码和解码的工作</p>

        <pre>
<code class="language-java">public class HelloServerAioHandler extends HelloAbsAioHandler implements ServerAioHandler&lt;Object, HelloPacket, Object&gt;
{
    /** 
     * 处理消息
     */
    @Override
    public Object handler(HelloPacket packet, ChannelContext&lt;Object, HelloPacket, Object&gt; channelContext) throws Exception
    {
        byte[] body = packet.getBody();
        if (body != null)
        {
            String str = new String(body, HelloPacket.CHARSET);
            System.out.println("收到消息：" + str);
 
            HelloPacket resppacket = new HelloPacket();
            resppacket.setBody(("收到了你的消息，你的消息是:" + str).getBytes(HelloPacket.CHARSET));
            Aio.send(channelContext, resppacket);
        }
        return null;
    }
}</code></pre>

        <ul>
            <li>
                <h3>启动程序</h3>
            </li>
        </ul>

        <pre>
<code class="language-java">public class HelloServerStarter
{
    //handler, 包括编码、解码、消息处理
    public static ServerAioHandler&lt;Object, HelloPacket, Object&gt; aioHandler = new HelloServerAioHandler();
     
    //事件监听器，可以为null，但建议自己实现该接口，可以参考showcase了解些接口
    public static ServerAioListener&lt;Object, HelloPacket, Object&gt; aioListener = null;
     
    //一组连接共用的上下文对象
    public static ServerGroupContext&lt;Object, HelloPacket, Object&gt; serverGroupContext = new ServerGroupContext&lt;&gt;(aioHandler, aioListener);
     
    //aioServer对象
    public static AioServer&lt;Object, HelloPacket, Object&gt; aioServer = new AioServer&lt;&gt;(serverGroupContext);
     
    //有时候需要绑定ip，不需要则null
    public static String serverIp = null;
     
    //监听的端口
    public static int serverPort = Const.PORT;
 
    /**
     * 启动程序入口
     */
    public static void main(String[] args) throws IOException
    {
        aioServer.start(serverIp, serverPort);
    }
}</code></pre>
    </li>
    <li>
        <h2>客户端代码</h2>

        <ul>
            <li>
                <h3>实现AioHandler</h3>
            </li>
        </ul>

        <p>注：此处继承自HelloAbsAioHandler，父类已经完成编码和解码的工作</p>

        <pre>
<code class="language-java">public class HelloClientAioHandler extends HelloAbsAioHandler implements ClientAioHandler&lt;Object, HelloPacket, Object&gt;
{
    /** 
     * 处理消息
     */
    @Override
    public Object handler(HelloPacket packet, ChannelContext&lt;Object, HelloPacket, Object&gt; channelContext) throws Exception
    {
        byte[] body = packet.getBody();
        if (body != null)
        {
            String str = new String(body, HelloPacket.CHARSET);
            System.out.println("收到消息：" + str);
        }
 
        return null;
    }
 
    private static HelloPacket heartbeatPacket = new HelloPacket();
 
    /** 
     * 此方法如果返回null，框架层面则不会发心跳；如果返回非null，框架层面会定时发本方法返回的消息包
     */
    @Override
    public HelloPacket heartbeatPacket()
    {
        return heartbeatPacket;
    }
}</code></pre>

        <ul>
            <li>
                <h3>启动程序</h3>
            </li>
        </ul>

        <pre>
<code class="language-java">public class HelloClientStarter
{
    //服务器节点
    public static Node serverNode = new Node("127.0.0.1", Const.PORT);
 
    //handler, 包括编码、解码、消息处理
    public static ClientAioHandler&lt;Object, HelloPacket, Object&gt; aioClientHandler = new HelloClientAioHandler();
     
    //事件监听器，可以为null，但建议自己实现该接口，可以参考showcase了解些接口
    public static ClientAioListener&lt;Object, HelloPacket, Object&gt; aioListener = null;
     
    //断链后自动连接的，不想自动连接请设为null
    private static ReconnConf&lt;Object, HelloPacket, Object&gt; reconnConf = new ReconnConf&lt;Object, HelloPacket, Object&gt;(5000L);
 
    //一组连接共用的上下文对象
    public static ClientGroupContext&lt;Object, HelloPacket, Object&gt; clientGroupContext = new ClientGroupContext&lt;&gt;(aioClientHandler, aioListener, reconnConf);
 
    public static AioClient&lt;Object, HelloPacket, Object&gt; aioClient = null;
    public static ClientChannelContext&lt;Object, HelloPacket, Object&gt; clientChannelContext = null;
 
    /**
     * 启动程序入口
     */
    public static void main(String[] args) throws Exception
    {
        aioClient = new AioClient&lt;&gt;(clientGroupContext);
        clientChannelContext = aioClient.connect(serverNode);
 
        //连上后，发条消息玩玩
        send();
    }
 
    private static void send() throws Exception
    {
        HelloPacket packet = new HelloPacket();
        packet.setBody("hello world".getBytes(HelloPacket.CHARSET));
        Aio.send(clientChannelContext, packet);
    }
}</code></pre>

        <p>&nbsp;</p>
    </li>
</ul>